package com.example.order.service.impl;

import com.alibaba.dubbo.config.annotation.Service;
import com.alibaba.fastjson.JSON;
import com.example.order.IPayService;
import com.example.order.constant.ShopCode;
import com.example.order.entity.Result;
import com.example.order.entity.TradeMqProducerTemp;
import com.example.order.entity.TradePay;
import com.example.order.entity.TradePayExample;
import com.example.order.exception.CastException;
import com.example.order.mapper.TradeMqProducerTempMapper;
import com.example.order.mapper.TradePayMapper;
import com.example.order.utils.IDWorker;
import io.netty.util.internal.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.util.Date;
@Slf4j
@Component
@Service(interfaceClass = IPayService.class)
public class PayServiceImpl implements IPayService{

    @Autowired
    private TradeMqProducerTempMapper mqProducerTempMapper;

    @Autowired
    private TradePayMapper tradePayMapper;

    @Autowired
    private IDWorker idWorker;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Value("${rocketmq.producer.group}")
    private String groupName;

    @Value("${mq.pay.topic}")
    private String topic;

    @Value("${mq.pay.tag}")
    private String tag;

    @Resource
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Override
    public Result createPayment(TradePay tradePay) {
        if(tradePay==null || tradePay.getOrderId()==null){
            CastException.cast(ShopCode.SHOP_REQUEST_PARAMETER_VALID);

        }
        //1.判断订单支付状态
        TradePayExample example = new TradePayExample();
        TradePayExample.Criteria criteria = example.createCriteria();
        criteria.andOrderIdEqualTo(tradePay.getOrderId());
        criteria.andIsPaidEqualTo(ShopCode.SHOP_PAYMENt_IS_PAID.getCode());
        int r = tradePayMapper.countByExample(example);
        if(r>0){
            CastException.cast(ShopCode.SHOP_PAYMENt_IS_PAID);
        }
        //2.设置订单状态为暂未支付
        tradePay.setIsPaid(ShopCode.SHOP_ORDER_PAY_STATUS_NO_PAY.getCode());
        //3.保存支付订单
        tradePay.setPayId(idWorker.nextId());
        tradePayMapper.insert(tradePay);
        return new Result(ShopCode.SHOP_SUCCESS.getSuccess(),ShopCode.SHOP_SUCCESS.getMessage());
    }

    //模拟第三方支付接口的回调
    public Result callbackPayment(TradePay tradePay){
        //1.判断用户支付状态
        log.info("支付回调");
        if(tradePay.getIsPaid().intValue()==ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY.getCode().intValue()){
        //2.更新支付订单状态为已经支付
            Long payId = tradePay.getPayId();
            TradePay pay = tradePayMapper.selectByPrimaryKey(payId);
        //3.判断支付订单是否存在
            if(pay==null){
                CastException.cast(ShopCode.SHOP_PAYMENt_NOT_PAID);
            }
            pay.setIsPaid(ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY.getCode());
            int r = tradePayMapper.updateByPrimaryKeySelective(pay);
            log.info("支付订单状态改为已支付");
            if(r==1){
                //3.创建支付成功的消息
                TradeMqProducerTemp tradeMqProducerTemp = new TradeMqProducerTemp();
                tradeMqProducerTemp.setId(String.valueOf(idWorker.nextId()));
                tradeMqProducerTemp.setGroupName(groupName);
                tradeMqProducerTemp.setMsgTopic(topic);
                tradeMqProducerTemp.setMsgTag(tag);
                tradeMqProducerTemp.setMsgKey(String.valueOf(tradePay.getPayId()));
                tradeMqProducerTemp.setMsgBody(JSON.toJSONString(tradePay));
                tradeMqProducerTemp.setCreateTime(new Date());
                //4.将消息持久化到数据库
                mqProducerTempMapper.insert(tradeMqProducerTemp);
                log.info("将支付成功消息持久化到数据库");
                //在线程池中处理
                threadPoolTaskExecutor.submit(()->{
                    //5.发送消息到MQ
                    SendResult result = sendMessage(topic,tag,String.valueOf(tradePay.getPayId()),JSON.toJSONString(tradePay));
                    if(result.getSendStatus().equals(SendStatus.SEND_OK)){
                        log.info("消息发送成功");
                        //6.等待结果，如果成功，删除掉消息
                        mqProducerTempMapper.deleteByPrimaryKey(tradeMqProducerTemp.getId());
                        log.info("删除持久化到数据库的消息");
                    }
                });
            }
        }else {
            CastException.cast(ShopCode.SHOP_PAYMENt_PAY_ERROR);
        }
        return null;
    }


    private SendResult sendMessage(String topic,String tag,String key,String body){
        SendResult sendResult = null;
        try {
            if(StringUtils.isEmpty(topic)){
                CastException.cast(ShopCode.SHOP_MQ_TOPIC_IS_Empty);
            }
            if(StringUtils.isEmpty(body)){
                CastException.cast(ShopCode.SHOP_MQ_MESSAGE_BODY_IS_EMPTY);
            }
            Message message = new Message(topic,tag,key,body.getBytes());
            sendResult  = rocketMQTemplate.getProducer().send(message);
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return sendResult;
    }
}
